#include "Reader.h"


#include "Persistence.h"

using namespace leveldb;

Reader::Reader(Stream * stream, uint64_t startSequenceNumber)
: stream_(stream)
, startSequenceNumber_(startSequenceNumber)
{
    this->open();
}
Reader::~Reader(){
    this->close();
}

leveldb::Status Reader::recover(){
    reader_ = stream_->createReader();

    uint64_t i = 0;
    for(; i < startSequenceNumber_ && reader_->hasNext(); ++i){
        reader_->next();
    }
    if(i < startSequenceNumber_){
        startSequenceNumber_ = i;
    }

    sequenceNumber_ = startSequenceNumber_;
    return Status::OK();
}

leveldb::Status Reader::open() {
    if(isOpen_ == false){
        assert(stream_ != nullptr);
        auto s = recover();
        if(!s.ok()){
            return s;
        }
        isOpen_ = true;
    }
    return leveldb::Status::OK();
}

leveldb::Status Reader::close(){
    if(isOpen_ == true){
        sequenceNumber_ = 0;
        delete reader_; reader_ = nullptr;
        stream_ = nullptr;

        isOpen_ = false;
    }
    return leveldb::Status::OK();
}

bool Reader::hasNext() const {
    assert(reader_ != nullptr);
    return reader_->hasNext();
}
void Reader::next() {
    assert(reader_ != nullptr);
    reader_->next();
    ++sequenceNumber_;
}

std::pair<uint64_t, const leveldb::Slice> Reader::current() {
    auto && p = this->_current();

    if(sequenceNumber_ < p.first)
        sequenceNumber_ = p.first;
    return p;	
}


std::pair<uint64_t, const leveldb::Slice> Reader::_current(){
    assert(reader_ != nullptr);

    KeyValueParser kvp(reader_->current());

    uint64_t key = 0;	
    leveldb::Slice result;
    if(kvp.keyValue(&key, &result)){
        return std::make_pair(key, result);
    }
    return std::pair(0, leveldb::Slice()); //TODO: ?? error ?????
}
